package ipl;
/*
 * YU LIU : 2012-11-30 @ Kochi
 */
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Hashtable;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class Scan {

	public static class FirstMapper extends
			Mapper<LongWritable, Text, IntWritable, IntPair> {

		private final static IntWritable one = new IntWritable(1);

		// private Text list;

		protected void setup(Context context) throws IOException,
				InterruptedException {
			// get
		}

		public void map(LongWritable key, Text list, Context context)
				throws IOException, InterruptedException {
			String[] eles = list.toString().trim().split(" ");
			int sum = 0;
			for (int i = 0; i < eles.length; i++) {
				sum += Integer.parseInt(eles[i]);
			}
			context.write(one, new IntPair((int) key.get(), sum));

		}
	}


	public static class ScanReducer extends
			Reducer<IntWritable, IntPair, Text, NullWritable> {
		String interSum = "";

		public void reduce(IntWritable key, Iterable<IntPair> values,
				Context context) throws IOException, InterruptedException {
			ArrayList<IntPair> sorted = new ArrayList<IntPair>();

			for (IntPair val : values) {
				IntPair tmp = new IntPair(val.first, val.second);
				sorted.add(tmp);
			}
			util.sortList(sorted); //sorting, for dis-parallel cases
			interSum = util.list2String(sorted);
			// sorted.
			context.write( new Text(util.list2String(sorted)),NullWritable.get());

		}

		protected void cleanup(Context context) throws IOException,
				InterruptedException {
			//context.getConfiguration().set("Scan_test", interSum);
		}
	}

//	public static class SecondMapper extends
//			Mapper<LongWritable, Text, Text, NullWritable> {
//
//		private ArrayList<Integer> scanList;
//		private Hashtable<Integer, Integer> keyIndex;
//
//		protected void setup(Context context) throws IOException,
//				InterruptedException {
//			Path[] paths = context.getLocalCacheFiles();
//			File f = new File(paths[0].toString());
//			//Files.readAllLines(new  java.nio.file.Path(paths[0].toString()),StandardCharsets.UTF_8);
//			Charset charset = StandardCharsets.UTF_8;
//			String partialSumList ="";
//			try (BufferedReader reader = Files.newBufferedReader(f.toPath(), charset)) {
//				
//			    String line = null;
//			    while ((line = reader.readLine()) != null) {
//			    	partialSumList += line;
//			    }
//			    partialSumList.trim();
//			} catch (IOException x) {
//			    System.err.format("IOException: %s%n", x);
//			}
//			
// //		String scanSum = FileSystem.get(context.getConfiguration()).open(new Path(paths[0].toString())).readLine().trim(); // not very good
////	scanSum = scanSum.split("\t")[1];
//
//			if(partialSumList.length() > 0){
//				keyIndex = new Hashtable<Integer, Integer>();
//				String[] tmp = partialSumList.split(";");
//				scanList = new ArrayList<Integer>();
//
//				scanList.add(0);
//				int sum = 0;
//				int counter = 0;
//				for (String val : tmp) {
//					if (val != null && val.length() > 0) {
//						int key = Integer.parseInt(val.split("=")[0]);
//						int parSum = Integer.parseInt(val.split("=")[1]);
//						keyIndex.put(key, counter);
//						sum += parSum;
//						scanList.add(sum); // add all of them to this list, from 0
//											// to last partial-sum
//					}
//					counter++;
//				}
//			}
//
//
//	
//		public void map(LongWritable key, Text list, Context context)
//				throws IOException, InterruptedException {
//			if (scanList.size() > 0) {
//				String[] eles = list.toString().trim().split(" ");
//				int idx = (int) key.get();
//				String newList = "";
//				// prescan
//				int sum = scanList.get(keyIndex.get(idx));
//				for (int i = 0; i < eles.length; i++) {
//					newList += sum + " ";
//					sum += Integer.parseInt(eles[i]);
//				}
//				context.write(new Text(newList), NullWritable.get());
//			}
//
//		}
//	}



//	public static class IDReducer extends
//			Reducer<LongWritable, Text, LongWritable, Text> {
//
//		public void reduce(LongWritable key, Iterable<Text> values,
//				Context context) throws IOException, InterruptedException {
//			ArrayList<IntPair> sorted = new ArrayList<IntPair>();
//
//			for (Text val : values) {
//				context.write(key, val);
//			}
//
//		}
//	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "scan1");
		job.setJarByClass(Scan.class);
		job.setMapperClass(FirstMapper.class);
		job.setReducerClass(ScanReducer.class);
		job.setNumReduceTasks(1);

		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(IntPair.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(Text.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		Path out = new Path(args[1] + "_int");
		FileOutputFormat.setOutputPath(job, out);
		Boolean ok = job.waitForCompletion(true);
		System.out.println("+++++++++++++++++++++++++++++++++");
		if (ok) {
			
			Configuration conf2 = new Configuration();
			FileSystem fs = FileSystem.get(conf2);
			Path partialSumList = new Path(out, "part-r-00000");
			
			//String scanSum = fs.open(partialSumList).readLine().trim(); // not very good
 			Path localCache = new Path(".partialSumList.txt");
     		fs.copyToLocalFile(partialSumList, localCache);
     		
     		Path cacheFile = new Path("scan/cache/partialSumList.txt");
     		if (fs.exists(cacheFile))
     			fs.delete(cacheFile, true);
     		fs.copyFromLocalFile(localCache,cacheFile);
//			scanSum = scanSum.split("\t")[1];
			
			DistributedCache.addCacheFile(cacheFile.toUri(), conf2);
			System.out.println(cacheFile.toUri());
			
			//conf2.set("partialSumListURI", localCache.toString());
			Job job2 = new Job(conf2, "scan2");
			//job2.addCacheFile(partialSumList.toUri());
			job2.setJarByClass(SecondMapper.class);
			job2.setMapperClass(SecondMapper.class);
			job2.setNumReduceTasks(0);
			job2.setMapOutputKeyClass(Text.class);
			job2.setMapOutputValueClass(NullWritable.class);
			// job2.setOutputKeyClass(NullWritable.class);
			// job2.setOutputValueClass(Text.class);
			job2.setInputFormatClass(TextInputFormat.class);
			job2.setOutputFormatClass(TextOutputFormat.class);
			FileInputFormat.addInputPath(job2, new Path(args[0]));
			FileOutputFormat.setOutputPath(job2, new Path(args[1]));
			int rst = job2.waitForCompletion(true) ? 0 : 1;
			fs.deleteOnExit(out);
			System.exit(rst);

		} else
			System.exit(1);

	}

}
